package com.young.core.manager.impl;


import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.young.core.dao.BrokerDao;
import com.young.core.domain.BrokerDO;
import com.young.core.manager.BrokerManager;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Component
public class BrokerManagerImpl implements BrokerManager {

    private static final String PREFIX_ID = "_B_IP_";

    private static final Cache<String, List<String>> B_CACHE = CacheBuilder.newBuilder().recordStats()
            .concurrencyLevel(Runtime.getRuntime().availableProcessors() + 2)
            .initialCapacity(4)
            .maximumSize(16)
            .expireAfterWrite(6, TimeUnit.SECONDS)
            .build();

    @Autowired
    private BrokerDao brokerDao;


    @Override
    public boolean keepAlive(String ip,String hostName) {
        boolean success = this.brokerDao.keepAlive(ip);
        if (!success) {
            BrokerDO broker = new BrokerDO();
            broker.setIp(ip);
            broker.setHostName(hostName);
            brokerDao.insert(broker);
        }
        return true;
    }

    @Override
    public void unregister(String ip) {
        this.brokerDao.delete(ip);
        B_CACHE.invalidateAll();
    }

    @Override
    public void expire(int ttl) {
        int rn = this.brokerDao.expire(ttl);
        if (rn > 0) {
            B_CACHE.invalidateAll();
        }
    }

    @Override
    public List<String> getCache() {
        try {
            return B_CACHE.get(PREFIX_ID, () -> get());
        } catch (CacheLoader.InvalidCacheLoadException | ExecutionException e) {
            return Arrays.asList();
        }
    }

    private List<String> get() {
        List<BrokerDO> brokers = this.brokerDao.getAll();
        return brokers.stream().map(b -> b.getIp()).distinct().collect(Collectors.toList());
    }

}
